package server

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"net/http"
	"sync"
	"time"

	v1 "user/api/ws/v1"
	"user/internal/data"

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
	"google.golang.org/protobuf/encoding/protojson"
)

// upgrader 用于将 HTTP 连接升级为 WebSocket 连接，允许所有来源的连接
var upgrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool { return true },
}

// WsServer 表示 WebSocket 服务器，管理所有的 WebSocket 连接和消息处理
type WsServer struct {
	connections map[string]*websocket.Conn // userID -> conn，存储每个用户的 WebSocket 连接
	lastActive  map[string]time.Time       // 新增，记录每个用户最后活跃时间
	mu          sync.RWMutex               // 读写锁，用于保护 connections 映射的并发访问
	Repo        *data.ChatRepo             // 消息存储仓库，用于消息的持久化
}

// NewWsServerWithRepo 创建一个新的 WebSocket 服务器实例，并初始化连接映射
func NewWsServerWithRepo(repo *data.ChatRepo, rabbitMQ *data.RabbitMQ) *WsServer {
	ws := &WsServer{
		connections: make(map[string]*websocket.Conn),
		lastActive:  make(map[string]time.Time),
		Repo:        repo,
	}
	// 启动 RabbitMQ 消费端
	ws.StartRabbitMQConsumer(rabbitMQ)
	return ws
}

// StartHeartbeatChecker 启动心跳检测 goroutine，定期清理超时连接
func (w *WsServer) StartHeartbeatChecker(timeout time.Duration) {
	go func() {
		for {
			time.Sleep(timeout / 2)
			now := time.Now()
			w.mu.Lock()
			for userID, t := range w.lastActive {
				if now.Sub(t) > timeout {
					if conn, ok := w.connections[userID]; ok {
						err := conn.Close()
						if err != nil {
							return
						}
						delete(w.connections, userID)
						delete(w.lastActive, userID)
						log.Printf("User %s connection closed due to heartbeat timeout", userID)
					}
				}
			}
			w.mu.Unlock()
		}
	}()
}

// StartPingAllClients 启动一个后台协程，定期向所有已连接的 WebSocket 客户端发送 Ping 帧，
// 用于检测客户端连接是否仍然活跃。
func (w *WsServer) StartPingAllClients() {
	// 启动一个独立的 goroutine 来执行定时 Ping 任务，避免阻塞主线程。
	go func() {
		// 进入无限循环，持续执行 Ping 操作
		for {
			// 每 10 秒执行一次 Ping 操作
			time.Sleep(10 * time.Second)
			// 加读锁，防止在遍历连接时其他协程修改 connections 映射
			w.mu.RLock()
			// 遍历所有已连接的客户端
			for userID, conn := range w.connections {
				if err := conn.WriteMessage(websocket.PingMessage, []byte("ping")); err != nil {
					log.Printf("Ping to %s failed: %v", userID, err)
				}
			}
			// 释放读锁，允许其他协程对 connections 映射进行读写操作
			w.mu.RUnlock()
		}
	}()
}

// HandleWS 处理 WebSocket 连接请求，将 HTTP 连接升级为 WebSocket 连接，并处理消息的读写操作
func (w *WsServer) HandleWS(ws http.ResponseWriter, r *http.Request) {
	// 将 HTTP 连接升级为 WebSocket 连接
	conn, err := upgrader.Upgrade(ws, r, nil)
	if err != nil {
		log.Printf("WebSocket upgrade error: %v", err)
		return
	}
	var userID string
	// 延迟执行，确保在函数返回时关闭连接并清理资源
	defer func() {
		log.Println("WebSocket connection closed")
		// 关闭 WebSocket 连接
		if err := conn.Close(); err != nil {
			log.Printf("Failed to close WebSocket connection: %v", err)
		}
		if userID != "" {
			// 锁定写锁，防止并发修改 connections 映射
			w.mu.Lock()
			// 从连接映射中删除该用户的连接
			delete(w.connections, userID)
			delete(w.lastActive, userID) // 断开时清理 lastActive
			// 解锁写锁
			w.mu.Unlock()
		}
	}()

	for {
		// 从 WebSocket 连接读取消息
		_, message, err := conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Printf("read error: %v", err)
			}
			break
		}
		log.Printf("Received message: %s", message)

		// 识别心跳包（假设心跳包格式为 {"type":"heartbeat"}）
		var base struct {
			Type string `json:"type"`
		}
		if err := json.Unmarshal(message, &base); err == nil && base.Type == "heartbeat" {
			w.mu.Lock()
			if userID != "" {
				w.lastActive[userID] = time.Now()
			}
			w.mu.Unlock()
			// 回复 pong
			pong := map[string]string{"type": "pong"}
			if err := conn.WriteMessage(websocket.TextMessage, mustJson(pong)); err != nil {
				log.Printf("Failed to send pong: %v", err)
			}
			continue
		}

		// 如果用户还未登录
		if userID == "" {
			var login v1.SendMessageRequest
			// 尝试将接收到的消息解析为登录请求
			if err := json.Unmarshal(message, &login); err == nil && login.SenderId != "" {
				userID = login.SenderId
				// 锁定写锁，防止并发修改 connections 映射
				w.mu.Lock()
				// 将用户的连接添加到连接映射中
				w.connections[userID] = conn
				w.lastActive[userID] = time.Now() // 登录时初始化活跃时间
				// 解锁写锁
				w.mu.Unlock()
				log.Printf("User %s connected", userID)
				continue
			}
			// 未登录直接丢弃消息
			continue
		}

		var req v1.SendMessageRequest
		// 尝试将接收到的消息解析为发送消息请求
		if err := json.Unmarshal(message, &req); err != nil {
			log.Printf("Invalid message format: %v", err)
			resp := map[string]interface{}{"error": "invalid_format", "message": err.Error()}
			// 向客户端发送消息格式错误的响应
			if err := conn.WriteMessage(websocket.TextMessage, mustJson(resp)); err != nil {
				log.Printf("Failed to send error response: %v", err)
			}
			continue
		}
		if req.ReceiverId == "" || req.Message == nil {
			log.Printf("Missing receiver_id or message body")
			resp := map[string]interface{}{"error": "missing_field", "message": "receiver_id or message body required"}
			// 向客户端发送缺少必要字段的响应
			if err := conn.WriteMessage(websocket.TextMessage, mustJson(resp)); err != nil {
				log.Printf("Failed to send error response: %v", err)
			}
			continue
		}

		// 生成唯一消息ID
		msgID := uuid.NewString()
		// 构造消息项
		item := v1.MessageItem{
			MessageId:  msgID,
			SenderId:   req.SenderId,
			ReceiverId: req.ReceiverId,
			Message:    req.Message,
			Timestamp:  time.Now().Unix(), // 服务器生成时间戳
		}

		// 消息持久化
		if w.Repo != nil {
			if err := w.Repo.SaveMessage(&item); err != nil {
				log.Printf("Failed to save message: %v", err)
			}
		}

		push, err := json.Marshal(&item)
		if err != nil {
			log.Printf("Failed to marshal message item: %v", err)
			continue
		}
		// 向接收者推送消息
		go func() {
			err := w.SendToUser(req.ReceiverId, push)

			if err != nil {
				log.Printf("Send to user %s failed: %v", req.ReceiverId, err)
				resp := map[string]interface{}{"error": "user_offline", "message": "user not online"}
				// 向发送者发送接收者不在线的响应
				if err := conn.WriteMessage(websocket.TextMessage, mustJson(resp)); err != nil {
					log.Printf("Failed to send user offline response: %v", err)
				}
			}

		}()

	}
}

// mustJson 用于快速将对象转为 JSON 字节切片，忽略可能的错误
func mustJson(v interface{}) []byte {
	b, _ := json.Marshal(v)
	return b
}

// ErrUserNotOnline 表示用户不在线的错误
var ErrUserNotOnline = errors.New("user is not online")

// SendToUser 向指定用户推送消息
func (s *WsServer) SendToUser(userID string, message []byte) error {
	// 使用读锁锁定连接映射，防止在读取连接时被其他 goroutine 修改
	s.mu.RLock()
	// 从连接映射中根据用户 ID 获取对应的 WebSocket 连接
	conn, ok := s.connections[userID]
	// 释放读锁，允许其他 goroutine 继续访问连接映射
	s.mu.RUnlock()

	// 如果没有找到对应的连接，说明用户不在线，返回错误
	if !ok {
		return ErrUserNotOnline // 用户不在线
	}

	// 向找到的 WebSocket 连接发送文本消息
	return conn.WriteMessage(websocket.TextMessage, message)
}

// StartRabbitMQConsumer 启动RabbitMQ消费端，收到消息后推送给在线用户
func (w *WsServer) StartRabbitMQConsumer(mq *data.RabbitMQ) {
	handler := func(msg *v1.MessageItem) {
		// 推送给在线用户
		w.mu.RLock()
		conn, ok := w.connections[msg.ReceiverId]
		w.mu.RUnlock()
		if ok {
			push, err := protojson.Marshal(msg)
			if err == nil {
				_ = conn.WriteMessage(1, push)
			}
		}
		// 持久化到 MongoDB
		if w.Repo != nil && w.Repo.Coll != nil {
			_, err := w.Repo.Coll.InsertOne(context.Background(), msg)
			if err != nil {
				log.Printf("Failed to persist message to MongoDB: %v", err)
			}
		}
	}
	if err := mq.ConsumeMessages(handler); err != nil {
		log.Printf("RabbitMQ consumer start error: %v", err)
	}
}
